package com.iot.mqtt;

import cn.hutool.extra.spring.SpringUtil;
import cn.hutool.json.JSONUtil;
import com.iot.ruler.caster.RulerPublisherIns;
import lombok.extern.slf4j.Slf4j;
import org.apache.ibatis.reflection.ArrayUtil;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
 * Mqtt辅助类
 * 使用流程：
 * step1：
 * 实现{@link org.springframework.context.ApplicationListener}接口，监听{@link MessageCallBackSubject}
 * step2：
 * 配置参数：
 * mqtt:
 * host: 124.222.244.193 //主机域名
 * port: 1883 // 端口号
 * username: admin // 账号
 * password: lzblzb123 //密码
 * topics: Data/#, Model/# , Config/#, Date_req // 连接后初始订阅主题
 * qos: 2, 2, 2, 2 //连接后订阅主题允许服务端使用的最大qos
 * autoConnect: true // 初始化后是否自动发起连接
 *
 * @author lzb
 */
@Slf4j
@Component
@Order(1)
public class MqttHelper implements MqttCallbackExtended, ApplicationRunner, DisposableBean {

    private final MqttConfig mqttConfig;
    /**
     * Mqtt异步对象
     */
    private MqttAsyncClient mqttClient;
    /**
     * Mqtt客户端配置参数
     */
    private MqttConnectOptions options;
    /**
     * mqtt回调接口
     */
    private MqttListener listener;

    public MqttHelper(MqttConfig mqttConfig) {
        this.mqttConfig = mqttConfig;
    }

    /**
     * 获取Bean实例
     */
    public static MqttHelper getInstance() {
        return SpringUtil.getBean(MqttHelper.class);
    }

    /**
     * 初始化客户端
     *
     * @param targetServer 目标mqtt地址
     * @param userName     用户名
     * @param password     密码
     */
    public void initMqtt(String targetServer, String userName, String password, boolean autoConnect) {
        try {
            mqttClient = new MqttAsyncClient(targetServer,
                    UUID.randomUUID().toString(),
                    new MemoryPersistence()
            );
            mqttClient.setCallback(this);
            options = new MqttConnectOptions();
            options.setUserName(userName);
            options.setPassword(password.toCharArray());
            options.setKeepAliveInterval(20);
            options.setMaxInflight(20);
            options.setAutomaticReconnect(false);
            options.setMaxReconnectDelay(10_000);
            options.setCleanSession(true);
            if (autoConnect) {
                connect();
            }
        } catch (MqttException e) {
            log.error("initMqtt: ", e);
        }
    }

    /**
     * 发送消息
     *
     * @param topicName 目标主题
     * @param msg       需要发送出去的消息
     * @param qos       QOS0： “至多一次”，消息发布完全依赖底层 TCP/IP 网络。会发生消息丢失或重复。
     *                  QOS1： “至少一次”，确保消息到达，但消息重复可能会发生。
     *                  QOS2： “只有一次”，确保消息到达一次
     */
    public void send(String topicName, String msg, int qos) {
        if (mqttClient != null && mqttClient.isConnected()) {
            MqttMessage message = new MqttMessage();
            message.setQos(qos);
            message.setPayload(msg.getBytes());
            try {
                mqttClient.publish(topicName, message);
            } catch (MqttException e) {
                log.error("消息发送失败:", e);
            }
        }
    }

    /**
     * 发送消息
     *
     * @param topicName 目标主题
     * @param msg       需要发送出去的消息
     */
    public void send(String topicName, String msg) {
        if (mqttClient != null && mqttClient.isConnected()) {
            MqttMessage message = new MqttMessage();
            message.setQos(0);
            message.setPayload(msg.getBytes());
            try {
                mqttClient.publish(topicName, message);
            } catch (MqttException e) {
                log.error("消息发送失败:", e);
            }
        }
    }

    /**
     * 发送消息
     *
     * @param topicName 需要发送出去的消息
     */
    public void send(String topicName, Object object) {
        if (mqttClient != null && mqttClient.isConnected()) {
            MqttMessage message = new MqttMessage();
            message.setQos(0);
            message.setPayload(JSONUtil.toJsonStr(object).getBytes());
            try {
                log.info("send: {}", topicName);
                mqttClient.publish(topicName, message);
            } catch (MqttException e) {
                log.error("消息发送失败:", e);
            }
        }
    }

    /**
     * 发起连接
     */
    public void connect() {
        if (mqttClient != null) {
            try {
                log.info("connect: 发起连接");
                mqttClient.connect(options, null, iMqttActionListener);
            } catch (MqttException e) {
                log.error("connect: ", e);
            }
        }
    }    /**
     * 连接状态回调
     */
    private final IMqttActionListener iMqttActionListener = new IMqttActionListener() {
        @Override
        public void onSuccess(IMqttToken asyncActionToken) {
            log.info("connected： " + asyncActionToken.getClient().getServerURI());
        }

        @Override
        public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
            log.error("onFailure:mqtt connect failed", exception);
            try {
                TimeUnit.SECONDS.sleep(30);
                connect();
            } catch (InterruptedException e) {
                log.error("mqtt retry connecting", e);
            }
        }
    };

    /**
     * 等待所有数据处理完毕后断开连接，等待时间最长不超过30s
     */
    public void disConnect() {
        if (mqttClient != null) {
            try {
                mqttClient.disconnect();
            } catch (MqttException e) {
                log.error("disConnect: ", e);
            }
        }
    }

    //MqttCallbackExtended接口方法-start
    @Override
    public void connectionLost(Throwable cause) {
        log.error("connectionLost", cause);
        if (listener != null) {
            listener.onConnectionLost(cause);
        }
        try {
            TimeUnit.SECONDS.sleep(30);
            connect();
        } catch (InterruptedException e) {
            log.error("mqtt retry connecting", e);
        }
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) {
        String messageStr = new String(message.getPayload(), StandardCharsets.UTF_8);
        int qos = message.getQos();
        if (listener != null) {
            listener.onMessage(topic, qos, messageStr);
        }
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        try {
            MqttMessage message = token.getMessage();
            String[] topics = token.getTopics();
            if (message != null) {
                String messageStr = new String(message.getPayload());
                if (listener != null) {
                    listener.onDeliveryComplete(topics, messageStr);
                }
            }
        } catch (MqttException e) {
            log.error("there was a problem completing retrieving the message: ", e);
        }
    }

    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        log.info("connectComplete:  reconnect=" + reconnect + ",serverURI=" + serverURI);
        if (listener != null) {
            listener.onConnected(MqttHelper.this, serverURI);
        }
    }

    /**
     * 设置回调
     * 连接状态、数据监听
     *
     * @param mqttListener 连接成功回调
     */
    public void setListener(MqttListener mqttListener) {
        this.listener = mqttListener;
    }
    //MqttCallbackExtended接口方法-end

    /**
     * 订阅服务器主题
     * 返回数据格式为topic:playload
     */
    public void subscribe(String topicName, int qos) {
        try {
            //订阅队列
            mqttClient.subscribe(topicName, qos);
        } catch (MqttException e) {
            log.error("subscribe: ", e);
        }
    }

    /**
     * 批量订阅服务器主题
     * 返回数据格式为topic:playload
     */
    public void subscribe(String[] topicName, int[] qos) {
        try {
            //订阅队列
            mqttClient.subscribe(topicName, qos);
            log.info("have subscribed topics:{}", ArrayUtil.toString(topicName));
            log.info("have subscribed qos:   {}", ArrayUtil.toString(qos));
            log.info("\n==============================================\n" +
                    "                   Mqtt Started!                  " +
                    "\n==============================================\n");
        } catch (MqttException e) {
            log.error("subscribe: ", e);
        }
    }

    @Override
    public void run(ApplicationArguments args) {
        log.info("\n==============================================\n" +
                "                SpringBoot Started                " +
                "\n==============================================\n");
        setListener(new MqttListener() {
            @Override
            public void onConnected(MqttHelper helper, String serverURI) {
                log.info("have connected to:" + serverURI);
                subscribe(mqttConfig.getTopics(), mqttConfig.getQos());
            }

            @Override
            public void onMessage(String topic, int qos, String message) {
                log.info("messageArrived: topic=" + topic + "，qos=" + qos);
                RulerPublisherIns.getInstance().publish(new MessageCallBackSubject(message, topic, qos));
            }

            @Override
            public void onConnectionLost(Throwable cause) {
                log.error("onConnectionLost", cause);
            }

            @Override
            public void onDeliveryComplete(String[] topics, String message) {
                log.info("topic:{},message:{}", ArrayUtil.toString(topics), message);
            }
        });
        initMqtt("tcp://" + mqttConfig.getHost() + ":" + mqttConfig.getPort(),
                mqttConfig.getUsername(), mqttConfig.getPassword(), mqttConfig.isAutoConnect());
    }

    @Override
    public void destroy() {
        log.info("\n==============================================\n" +
                "              SpringBoot preDestroy               " +
                "\n==============================================\n");
        disConnect();
    }




}
